package com.nlp.server.utils;

import com.nlp.common.core.exception.ServiceException;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * sse连接工具类
 * @author shiliuyinzhen
 */
public class SseEmitterUtils {

    /**
     * 容器，保存连接，用于输出返回
     */
    private static Map<Long, SseEmitter> sseMap = new ConcurrentHashMap<>();

    /**
     * 创建sse连接
     * @param userId
     * @return
     */
    public static SseEmitter addSseEmitter(Long userId){
        SseEmitter sseEmitter = sseMap.get(userId);
        if(sseEmitter == null){
            sseEmitter = new SseEmitter(30000L); //超时时间30s
            sseEmitter.onTimeout(() -> {
                sseMap.remove(userId);
            });
            sseEmitter.onError(throwable -> {
                sseMap.remove(userId);
            });
            SseEmitter finalSseEmitter = sseEmitter;
            sseEmitter.onCompletion(() -> {
                sseMap.put(userId, finalSseEmitter);
            });
        }
        try {
            sseEmitter.send(SseEmitter.event()
                    .id("200")
                    .name("test")
                    .data("连接成功"));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return sseEmitter;
    }

    /***
     * 关闭sse连接
     * @param userId
     * @return
     */
    public static boolean closeSseEmitter(Long userId) {
        SseEmitter sseEmitter = sseMap.get(userId);
        if (sseEmitter == null) {
            return true;
        }
        try {
            sseEmitter.complete();
            sseMap.remove(userId);
            return true;
        } catch (Exception e) {
            throw new ServiceException("关闭SSE连接失败,userId="+userId);
        }
    }

    /***
     * 给单个用户发消息
     * @param userId
     * @param msg
     * @return
     */
    public static boolean sendMsgToSingleClient(Long userId,Object msg) {
        SseEmitter sseEmitter = sseMap.get(userId);
        if (sseEmitter == null) {
            return false;
        }
        try {
            sseEmitter.send(SseEmitter.event()
                    .id("300")
                    .name("check-alive")
                    .data(msg));
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    /***
     * 检测客户端连接状态
     * @param sseEmitter
     * @return true代表还连接, false代表失去连接
     */
    public static boolean checkSseConnectAlive(SseEmitter sseEmitter) {
        if (sseEmitter == null) {
            return false;
        }
        // 返回true代表还连接, 返回false代表失去连接
        return !(Boolean) getField(sseEmitter,sseEmitter.getClass(), "sendFailed") &&
                !(Boolean) getField(sseEmitter,sseEmitter.getClass(), "complete");
    }

    /**
     * 通过反射,获取到sseEmitter类的属性判断是否存活
     * @param obj
     * @param clazz
     * @param fieldName
     * @return
     */
    public static Object getField(Object obj, Class<?> clazz, String fieldName) {
        for (; clazz != Object.class; clazz = clazz.getSuperclass()) {
            try {
                Field field;
                field = clazz.getDeclaredField(fieldName);
                field.setAccessible(true);
                return field.get(obj);
            } catch (Exception e) {

            }
        }
        return null;
    }

    /**
     * 获取Sse连接
     * @param userId
     * @return
     */
    public static SseEmitter get(Long userId) {
        return sseMap.get(userId);
    }

    /***
     * 给所有客户端发送检测消息
     * @param msg
     */
    public void sendMsgToAllClient(Object msg) {
        if (sseMap != null && !sseMap.isEmpty()) {
            for (Long userId : sseMap.keySet()) {
                // 发送检测消息
                sendMsgToSingleClient(userId,msg);
                // 判断客户端是否能接收到消息
                boolean isAlive = checkSseConnectAlive(sseMap.get(userId));
                //如果不能接收到消息
                if (!isAlive) {
                    // 断开连接，同时移除sse连接
                    closeSseEmitter(userId);
                }
            }
        }
    }

    /***
     * 定时判断所有客户端状态
     */
    @Async("sseExecutorService")
    @Scheduled(fixedDelayString = "${sse.checkAlive}", fixedDelay = 5, timeUnit = TimeUnit.MINUTES) // 默认5分钟
    public void checkAlive() {
        sendMsgToAllClient("CHECK_ALIVE");
    }

}
